﻿using Furion.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using OnWork.DataCenter.BLL.DataCenter;
using OnWork.DataCenter.DAL.DataCenter;
using System.Collections.Concurrent;
using System.Data;
using System.Text;

namespace OnWork.DataCenter.BLL.Standard.HostedService
{
    /// <summary>
    /// MQTT 后台服务 https://mlog.club/article/1137004
    /// </summary>
    public class MqttHostedService : IHostedService, IDisposable, ISingleton
    {
        private static ILogger<MqttHostedService> _logger;

        /// <summary>
        /// 客户端列表
        /// </summary>
        private static ConcurrentDictionary<long, IManagedMqttClient> DictClients = new ConcurrentDictionary<long, IManagedMqttClient>();

        private static System.Timers.Timer timer = new System.Timers.Timer();

        /// <summary>
        /// MQtt 发布消息
        /// </summary>
        /// <param name="dSMqtt">对象</param>
        /// <param name="topic">主题</param>
        /// <param name="content">内容</param>
        /// <returns></returns>
        public static async Task MqttPublish(DSMqtt dSMqtt, string topic, string content)
        {
            try
            {
                IManagedMqttClient managedMqttClient = null;

                if (DictClients.ContainsKey(dSMqtt.ID))
                {
                    managedMqttClient = DictClients[dSMqtt.ID];
                }
                else
                {
                    managedMqttClient = await MqttClientAdd(dSMqtt);
                }

                if (managedMqttClient != null && managedMqttClient.IsConnected)
                {
                    _logger.LogDebug($"发送信息{dSMqtt.Name}{dSMqtt.BrokerAddress}主题{topic}");
                  await managedMqttClient.EnqueueAsync(topic, content, MqttQualityOfServiceLevel.AtMostOnce, true);
                    //await managedMqttClient.EnqueueAsync(new MqttApplicationMessage()
                    //{
                    //    Topic = topic,
                    //    //Payload = Encoding.UTF8.GetBytes(content),
                    //    PayloadSegment = Encoding.UTF8.GetBytes(content),
                    //    // ContentType = "application/json",
                    //    QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
                    //    Retain = true, //设置为首次连接发送内容
                    //});
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex.ToString());
                throw;
            }
        }

        /// <summary>
        /// 数据查询
        /// </summary>
        private static DSMqttBLL _dSMqttBLL;

        /// <summary>
        /// 数据库日志
        /// </summary>
        private static DSMqttLogBLL _dSMqttLogBLL;

        public MqttHostedService(ILogger<MqttHostedService> logger, DSMqttLogBLL dSMqttLogBLL)
        {
            _logger = logger;
            _dSMqttLogBLL = dSMqttLogBLL;
            _dSMqttBLL = new DSMqttBLL();
        }

        /// <summary>
        /// 更新 MQTT的方法
        /// </summary>
        /// <returns></returns>
        public async Task UpdateMqtt()
        {
            await CloseMqttClientAsync();
            await LoadMqttClientAsync();
        }

        /// <summary>
        /// 加载异步方法
        /// </summary>
        /// <returns></returns>
        private async Task LoadMqttClientAsync()
        {
            DictClients.Clear(); //清空

            //获取启用的数据
            var mqts = await _dSMqttBLL.QueryEnableMqttAsync();

            if (mqts != null)
            {
                foreach (var mqtt in mqts)
                {
                    //新增测试功能
                    await MqttClientAdd(mqtt);
                }
            }
        }

        /// <summary>
        /// 关闭请求
        /// </summary>
        /// <returns></returns>
        private async Task CloseMqttClientAsync()
        {
            foreach (var item in DictClients.Keys)
            {
                var client = DictClients[item];
                await client.InternalClient.DisconnectAsync();
                await client.StopAsync();
            }
            DictClients.Clear(); //清空
        }

        /// <summary>
        /// 添加MQTT数据
        /// </summary>
        /// <param name="dSMqtt"></param>
        /// <returns></returns>
        private static async Task<IManagedMqttClient> MqttClientAdd(DSMqtt dSMqtt)
        {
            IManagedMqttClient managedMqttClient = null;

            if (DictClients.ContainsKey(dSMqtt.ID))
            {
                managedMqttClient = DictClients[dSMqtt.ID];
            }

            if (managedMqttClient == null)
            {
                var mqttFactory = new MqttFactory();
                managedMqttClient = mqttFactory.CreateManagedMqttClient();
            }
            var mqttClientOptions = new MqttClientOptionsBuilder()
                .WithTcpServer(dSMqtt.BrokerAddress, dSMqtt.BrokerPort)
                .WithClientId(dSMqtt.ClientID)
                .WithCredentials(dSMqtt.UserName, dSMqtt.Password)
                .Build();

            var managedMqttClientOptions = new ManagedMqttClientOptionsBuilder()
                .WithClientOptions(mqttClientOptions)
                .Build();

            //注册接收消息
            if (true)
            {
                //managedMqttClient.ConnectingFailedAsync += MqttClient_ConnectingFailedAsync;
                //managedMqttClient.DisconnectedAsync += MqttClient_DisconnectedAsync;
                //managedMqttClient.ConnectedAsync += MqttClient_ConnectedAsync;
                //managedMqttClient.ApplicationMessageReceivedAsync += MqttClientMessageReceivedHandler;

                //获取到订阅的消息
                managedMqttClient.ApplicationMessageReceivedAsync += async (arg) =>
                {
                    await MqttClientMessageReceivedHandler(arg, managedMqttClient, dSMqtt);
                    //await Task.CompletedTask;
                };

                //断开连接
                managedMqttClient.DisconnectedAsync += async (arg) =>
                {
                    await MqttClient_DisconnectedAsync(arg, managedMqttClient, dSMqtt);
                    await Task.CompletedTask;
                };
                //连接成功
                managedMqttClient.ConnectedAsync += async (arg) =>
                {
                    await MqttClient_ConnectedAsync(arg, managedMqttClient, dSMqtt);
                    await Task.CompletedTask;
                };
            }

            if (managedMqttClient.IsStarted)
            {
                await managedMqttClient.StopAsync();
            }

            await managedMqttClient.StartAsync(managedMqttClientOptions);
            DictClients[dSMqtt.ID] = managedMqttClient;
            _logger.LogDebug($"新增MQTT连接{dSMqtt.Name}{dSMqtt.BrokerAddress}");
            return managedMqttClient;
        }

        /// <summary>
        /// 获取到订阅的消息
        /// </summary>
        /// <param name="arg"></param>
        /// <param name="managedMqttClient">请求客户端</param>
        /// <param name="dSMqtt">配置</param>
        /// <returns></returns>
        private static async Task MqttClientMessageReceivedHandler(MqttApplicationMessageReceivedEventArgs arg, IManagedMqttClient managedMqttClient, DSMqtt dSMqtt)
        {
            string content = Encoding.UTF8.GetString(arg.ApplicationMessage.PayloadSegment);

            //DSMqttLog log = new DSMqttLog()
            //{
            //    ClientID = arg.ClientId,
            //    MqttID = dSMqtt.ID,
            //    Type = "接收订阅",
            //    Topic = arg.ApplicationMessage.Topic,
            //    Content = content,
            //    Password = dSMqtt.Password,
            //    BrokerPort = dSMqtt.BrokerPort,
            //    UserName = dSMqtt.UserName,
            //};
            //_dSMqttLogBLL.AddSplitTable(log);

            //获取订阅的主题
            var tops = dSMqtt.Topics.Where(x => x.Topic == arg.ApplicationMessage.Topic).ToList();

            //Console.WriteLine();
            //string msg = $"{DateTimeOffset.Now}:{arg.ApplicationMessage.Topic}:{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}";
            //_logger.LogInformation(msg);
            //处理后台数据
            await ProcessData(dSMqtt, arg.ApplicationMessage.Topic, content);
        }

        /// <summary>
        /// 处理数据
        /// </summary>
        /// <param name="dSMqtt"></param>
        /// <param name="topic"></param>
        /// <returns></returns>
        private static async Task ProcessData(DSMqtt dSMqtt, string topic, string content)
        {
            var list = await _dSMqttBLL.GetDataStandAsync(dSMqtt, topic);

            if (list != null && list.Count > 0)
            {
                foreach (var item in list)
                {
                    await ProcessStandards(item, content);
                }
            }
        }

        /// <summary>
        /// 处理标准
        /// </summary>
        /// <param name="dataStandards"></param>
        /// <param name="content"></param>
        /// <returns></returns>
        private static async Task ProcessStandards(MqttProcessStandardDto data, string content)
        {
            //像队列插入内容
            StandardsQueue.Enqueue(new StandardsQueueModel() { Standard = data, Content = content });
            await Task.CompletedTask;
            //await Task.Run(() =>
            //{
            //    StandardConvert.ProcessTreeStandards(data.StandardsTrees, content);
            //    //StandardConvert.ProcessStandards(dataStandards, content); //处理当前数据
            //}
            //);
        }

        //并发队列缓存
        private static ConcurrentQueue<StandardsQueueModel> StandardsQueue = new ConcurrentQueue<StandardsQueueModel>();

        /// <summary>
        /// 队列任务处理
        /// </summary>
        /// <param name="queue"></param>
        /// <param name="name"></param>
        /// <param name="token"></param>
        /// <returns></returns>
        static async Task TaskProcessor(ConcurrentQueue<StandardsQueueModel> queue, string name, CancellationToken token)
        {
            try
            {
                _logger.LogInformation(name + "任务启动!");
                StandardsQueueModel workItem;
                bool dequeueSuccesful = false;
                //await GetRandomDelay();
                do
                {
                    dequeueSuccesful = queue.TryDequeue(out workItem);
                    if (dequeueSuccesful)
                    {
                        //Console.WriteLine("task {0} has been processed by {1}", workItem.Id, name);
                        StandardConvert.ProcessTreeStandards(workItem.Standard.StandardsTrees, workItem.Content);//处理流程
                    }
                    else
                    {
                        Thread.Sleep(20); //如果没有内容则睡眠100ms 减少cpu使用
                    }
                    // await GetRandomDelay();
                }
                while (!token.IsCancellationRequested);

                await Task.CompletedTask;
                _logger.LogInformation(name + "任务停止!");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex.ToString());
                throw;
            }
        }

        /// <summary>
        /// 任务取消对象
        /// </summary>
        private static CancellationTokenSource? taskCancel;

        /// <summary>
        /// 执行队列任务
        /// </summary>
        /// <returns></returns>
        private static void RunProgram()
        {
            _logger.LogInformation("启动队列任务!");
            var taskQueue = StandardsQueue;
            var cts = new CancellationTokenSource();
            taskCancel = cts;
            //同时启动四个任务处理队列中的任务
            Task[] processors = new Task[8];
            for (int i = 1; i <= 8; i++)
            {
                string processId = i.ToString();
                processors[i - 1] = Task.Run(
                     () => TaskProcessor(taskQueue, "Processor " + processId, cts.Token)
                    );
            }
            //await Task.CompletedTask;
            _logger.LogInformation("启动队列任务完成!");

            ////新开线程后台执行
            //await Task.Factory.StartNew(async () =>
            //{
            //    var taskQueue = StandardsQueue;
            //    var cts = new CancellationTokenSource();
            //    taskCancel = cts;
            //    //同时启动四个任务处理队列中的任务
            //    Task[] processors = new Task[8];
            //    for (int i = 1; i <= 8; i++)
            //    {
            //        string processId = i.ToString();
            //        processors[i - 1] = await Task.Run(
            //            async () =>  TaskProcessor(taskQueue, "Processor " + processId, cts.Token)
            //            );
            //    }
            //    //await taskSource;
            //    //向任务发送取消信号
            //    //cts.CancelAfter(TimeSpan.FromSeconds(2));
            //    //await Task.WhenAll(processors);
            //});
        }

        /// <summary>
        /// 停止任务
        /// </summary>
        /// <returns></returns>
        private static void StopProgram()
        {
            if (taskCancel != null)
            {
                taskCancel.CancelAfter(TimeSpan.FromSeconds(1));
            }
        }

        /// <summary>
        /// 连接成功
        /// </summary>
        /// <param name="arg"></param>
        /// <param name="managedMqttClient">请求客户端</param>
        /// <param name="dSMqtt">配置</param>
        /// <returns></returns>
        private static async Task MqttClient_ConnectedAsync(MqttClientConnectedEventArgs arg, IManagedMqttClient managedMqttClient, DSMqtt dSMqtt)
        {
            DSMqttLog log = new DSMqttLog()
            {
                ClientID = dSMqtt.ClientID,
                MqttID = dSMqtt.ID,
                Type = "连接成功",
                Topic = "",
                Content = "",
                Password = dSMqtt.Password,
                BrokerPort = dSMqtt.BrokerPort,
                UserName = dSMqtt.UserName,
            };
            _dSMqttLogBLL.AddSplitTable(log);

            _logger.LogInformation($"MqttMqtt客户端连接成功!");
            //连接成功订阅主题
            if (dSMqtt.Topics != null)
            {
                var topics = dSMqtt.Topics.Select(x => x.Topic).Distinct().ToArray();
                await SubscribeTopics(managedMqttClient, topics);
            }
        }

        /// <summary>
        /// 断开连接
        /// </summary>
        /// <param name="arg"></param>
        /// <param name="managedMqttClient">请求客户端</param>
        /// <param name="dSMqtt">配置</param>
        /// <returns></returns>
        private static async Task MqttClient_DisconnectedAsync(MqttClientDisconnectedEventArgs arg, IManagedMqttClient managedMqttClient, DSMqtt dSMqtt)
        {
            DSMqttLog log = new DSMqttLog()
            {
                ClientID = dSMqtt.ClientID,
                MqttID = dSMqtt.ID,
                Type = "连接断开",
                Topic = "",
                Content = "",
                Password = dSMqtt.Password,
                BrokerPort = dSMqtt.BrokerPort,
                UserName = dSMqtt.UserName,
            };
            _dSMqttLogBLL.AddSplitTable(log);
            _logger.LogInformation($"Mqtt断开连接!" + arg.Exception);
            await Task.CompletedTask;
        }

        private Task MqttClient_ConnectingFailedAsync(ConnectingFailedEventArgs arg)
        {
            _logger.LogInformation($"Mqtt客户端连接失败!" + arg.Exception?.Message);
            return Task.CompletedTask;
        }

        /// <summary>
        /// 订阅主题
        /// </summary>
        /// <returns></returns>
        private static async Task SubscribeTopics(IManagedMqttClient managedMqttClient, string[] topics)
        {
            if (managedMqttClient == null || managedMqttClient.IsConnected == false)
            {
                _logger.LogInformation($"未连接请先连接");
                return;
            }

            MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;

            var mqttTopicFilters = new List<MqttTopicFilter>();

            foreach (var topic in topics)
            {
                mqttTopicFilters.Add(new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
            }
            await managedMqttClient.SubscribeAsync(mqttTopicFilters);
        }

        public void Dispose()
        {
            CloseMqttClientAsync();
        }

        public async Task StartAsync(CancellationToken cancellationToken)
        {
            //定时执行检测任务
            timer.Elapsed += new System.Timers.ElapsedEventHandler(OnTmrTrg);
            timer.Interval = 3000;
            //timer.AutoReset = true;
            //timer.Start();
            _logger.LogInformation("启动MQTT后台服务.");
            await LoadMqttClientAsync();
            RunProgram(); //启动后台队列
        }

        private void OnTmrTrg(object sender, System.Timers.ElapsedEventArgs e)
        {
            var count = StandardsQueue.Count();
            _logger.LogInformation($"当前MQTT队列总数:{StandardsQueue.Count}");
            int threshold = 1000;
            if (count > threshold)
            {
                StandardsQueue.Clear();
                _logger.LogInformation($"超过阈值{threshold}清空队列!");

                //停止任务
                StopProgram();
                RunProgram();
            }
            //WriteInfo("tmr loop:  " + Thread.CurrentThread.ManagedThreadId);
            //lock (obj)
            //{
            //    WriteInfo("tmr is running :" + Thread.CurrentThread.ManagedThreadId);
            //    for (int i = 0; i < 10; i++)
            //    {
            //        Thread.Sleep(1000);
            //        WriteInfo("Count: " + i + ": " + Thread.CurrentThread.ManagedThreadId);
            //    }
            //}
        }

        public async Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("结束MQTT后台服务.");
            await CloseMqttClientAsync();
        }
    }

    /// <summary>
    /// 队列对象
    /// </summary>
    public class StandardsQueueModel
    {
        public MqttProcessStandardDto Standard { get; set; }
        public string Content { get; set; }
    }
}